package learnjava.concur.chapter8;

import lombok.extern.slf4j.Slf4j;

import java.util.ArrayDeque;
import java.util.Deque;
import java.util.HashSet;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

@Slf4j
public class TestPool {

    public static void main(String[] args) {

        ThreadPool tp = new ThreadPool(5,2,1000,TimeUnit.MILLISECONDS, (queue,task)->{
            //任务队列满了以后的策略：
            //1）死等
//            queue.put(task);
            //2）带超时等待
             queue.put(task,500,TimeUnit.MILLISECONDS);
            //3）调用者放弃任务
//            System.out.println("放弃执行"+task);
            //4）调用者抛出异常
//            throw new RuntimeException("执行失败"+task);
            //5）调用者自己执行任务
//            task.run();
        });
        for (int i = 0; i < 10; i++) {
            int j = i;
            tp.execute(()->{
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                log.info("{}",j);
            });
        }
    }
}

//自定义函数式接口，处理任务队列满时的策略
@FunctionalInterface
interface RejectPolicy<T>{
    void reject(BlockingQueue<T> queue, T task);
}

@Slf4j
class ThreadPool {

    //任务队列
    private BlockingQueue<Runnable> taskQueue;

    //线程集合
    private HashSet<Worker> workers = new HashSet();

    //核心线程数
    private int coreSize;

    //线程空闲时间
    private long idleTime;
    private TimeUnit timeUnit;

    //拒绝策略
    RejectPolicy<Runnable> rejectPolicy;


    public ThreadPool(int queueCapacity, int coreSize, long idleTime, TimeUnit timeUnit, RejectPolicy<Runnable> rejectPolicy) {
        this.taskQueue = new BlockingQueue<>(queueCapacity);
        this.coreSize = coreSize;
        this.idleTime = idleTime;
        this.timeUnit = timeUnit;
        this.rejectPolicy = rejectPolicy;
    }

    public void execute(Runnable task){
        //为当前任务task安排一个线程处理它
        //如果当前线程数小于核心线程数，继续创建线程处理task
        //如果已经达到核心线程数，再创建线程也没有多余的cpu处理它，所以把task放入任务队列等待即可
        synchronized (workers){
            if(workers.size()<coreSize){
                Worker worker = new Worker(task);
                log.info("新增worker{}",worker);
                workers.add(worker);
                worker.start();
            }else {
//                taskQueue.put(task);
//                System.out.println("加入任务队列");
                /**
                 * 任务队列满了以后的策略：
                 * 1）死等
                 * 2）带超时等待
                 * 3）调用者放弃任务
                 * 4）调用者抛出异常
                 * 5）调用者自己执行任务
                 * 这么多策略，不如写个接口让调用者自已决定
                 */
                taskQueue.tryPut(rejectPolicy,task);
            }
        }
    }


    class Worker extends Thread{
        Runnable task;

        public Worker(Runnable task) {
            this.task = task;
        }

        @Override
        public void run() {
            while (task!=null||(task=taskQueue.take(idleTime,timeUnit))!=null){
                log.info("正在执行任务{}",task);
                task.run();
                task = null; //释放任务
            }
            synchronized (workers){
                log.info("正在移除worker{}",this);
                workers.remove(this);
            }
        }
    }
}

@Slf4j
class BlockingQueue<T> {
    //1，任务队列，ArrayDeque是一个循环队列
    private Deque<T> deque = new ArrayDeque<>();

    //2，锁
    private ReentrantLock lock = new ReentrantLock();

    //3，生产者条件变量
    private Condition fullWaiSet = lock.newCondition();

    //4，消费者条件变量
    private Condition emptyWaitSet = lock.newCondition();

    //5，容量
    private int capacity;

    public BlockingQueue(int capacity) {
        this.capacity = capacity;
    }

    //带超时的阻塞获取
    public T take(long timeout, TimeUnit unit){
        try {
            lock.lock();
            long nanos = unit.toNanos(timeout); //同义转换为纳秒
            while (deque.size()==0){
                try {
                    if(nanos<=0) return null;
                    //awaitNanos方法可以自动返回剩余等待时间
                    nanos = emptyWaitSet.awaitNanos(nanos);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            //deque不空了
            T e = deque.pollFirst();
            fullWaiSet.signal();
            return e;
        }finally {
            lock.unlock();
        }
    }

    //阻塞获取
    public T take(){
        try {
            lock.lock();
            while (deque.size()==0){
                try {
                    emptyWaitSet.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            //deque不空了
            T e = deque.pollFirst();
            fullWaiSet.signal();
            return e;
        }finally {
            lock.unlock();
        }
    }

    //阻塞添加
    public void put(T element){
        try {
            lock.lock();
            while (capacity==deque.size()){
                try {
                    fullWaiSet.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            deque.offerLast(element);
        }finally {
            lock.unlock();
        }
    }

    //带超时的阻塞添加
    public boolean put(T element, long timeout, TimeUnit unit){
        try {
            lock.lock();
            long nanos = unit.toNanos(timeout);
            while (capacity==deque.size()){
                try {
                    log.info("等待加入任务队列{}",element);
                    if(nanos<=0) return false;
                    nanos = fullWaiSet.awaitNanos(nanos);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            deque.offerLast(element);
            return true;
        }finally {
            lock.unlock();
        }
    }

    public void tryPut(RejectPolicy<T> rejectPolicy, T task){
        try {
            lock.lock();
            if(deque.size()<capacity){
                log.info("加入任务队列{}",task);
                deque.offerLast(task);
            }else { //满了
                rejectPolicy.reject(this,task);
            }
        }finally {
            lock.unlock();
        }
    }


    //获取大小
    public int getSize(){
        try {
            lock.lock();
            return deque.size();
        }finally {
            lock.unlock();
        }
    }

}
